package org.apache.lucene.distributed;

/**
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements. See the NOTICE file distributed with this
 * work for additional information regarding copyright ownership. The ASF
 * licenses this file to You under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * 
 * http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.Serializable;
import java.lang.ref.Reference;
import java.lang.ref.ReferenceQueue;
import java.lang.ref.SoftReference;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.net.ConnectException;
import java.net.InetSocketAddress;
import java.net.SocketTimeoutException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.IdentityHashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;

import javax.net.SocketFactory;

import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.ipc.Client;
import org.apache.lucene.distributed.NameService.Lease;
import org.apache.lucene.distributed.RMIClasses.ClassAccept;
import org.apache.lucene.distributed.RMIServerSession.RMIServerObjectOutputStream;

/*
 * Base class for performing RMI over Hadoop RPC. Creates a server and allows
 * for creation of proxy clients bound to a specific service. A service is a
 * remote class defined by an interface.
 * 
 * A default service is ClassService which allows the loading of class bytes.
 * This is used during the deserialization of a class on the Lucene server to
 * dynamically load classes that are not a part of the core Lucene library such
 * as those that subclass Query or Filter.
 * 
 * The RMI.Server.addService(String name, Object service) method allows defining
 * a service by name with the server. Like Sun's RMI, service interface need to
 * extends java.rmi.Remote. This is used for passing references to remote
 * services in remote method calls.
 * 
 */
// TODO: incoming call to server can set threadlocal of caller
// TODO: settings in hadoop for socket connections num tries and timeout need
// adjustment
// TODO: threadlocal on call of configuration desired
// TODO: need to generate jar file of classes referenced by serialized object,
// create md5
// hash of the classes and their serialuid
public class RMI {
  private static Map<SocketFactory,Client> CLIENTS = new HashMap<SocketFactory,Client>();
  Map<ProxyKey,RMIClient> rmiClientMap = new HashMap<ProxyKey,RMIClient>();
  Map<Object,RMIClient> objRmiClientMap = new IdentityHashMap<Object,RMIClient>();
  Map<Object,List<ProxyEventListener>> proxyEventListeners = new IdentityHashMap<Object,List<ProxyEventListener>>();
  Server server;
  SocketFactory socketFactory;
  String localhost;
  int port;
  ClassAccept classAccept;
  Configuration clientConf = new Configuration();
  ReferenceQueue proxyRefQueue = new ReferenceQueue();
  public static long LEASE_DURATION = 1000 * 30;
  private ScheduledExecutorService ses;
  int clientSocketTimeout = 1000 * 10;
  RMISecurity rmiSecurity;
  File tempDir;
  String sessionId = UUID.randomUUID().toString();

  public RMI(String bindAddress, int port, int numHandlers, ClassAccept classAccept) throws IOException {
    this(bindAddress, port, numHandlers, classAccept, null);
  }

  public RMI(String bindAddress, int port, int numHandlers, ClassAccept classAccept, RMISecurity rmiSecurity) throws IOException {
    socketFactory = SocketFactory.getDefault();
    this.port = port;
    localhost = bindAddress;// InetAddress.getLocalHost().getHostAddress();
    this.classAccept = classAccept;
    this.rmiSecurity = rmiSecurity;
    Configuration conf = new Configuration();
    server = new Server(conf, bindAddress, port, numHandlers, false);
    ses = Executors.newSingleThreadScheduledExecutor();
    clientConf.set("ipc.client.connect.max.retries", Integer.toString(3));
    ses.scheduleWithFixedDelay(new TimerTask(), LEASE_DURATION / 2, LEASE_DURATION / 2, TimeUnit.MILLISECONDS);
  }

  public InetSocketAddress getInetSocketAddress() {
    return new InetSocketAddress(localhost, port);
  }

  public synchronized void removeRMIClient(ProxyKey proxyKey) {
    RMIClient c = rmiClientMap.remove(proxyKey);
    if (c != null) {
      objRmiClientMap.remove(c.proxy);
      proxyEventListeners.remove(c.proxy);
    }
  }

  public static interface ProxyEventListener {
    public void proxyRemoved(Object proxy);
  }

  public void addProxyEventListener(Object proxy, ProxyEventListener pel) {
    List<ProxyEventListener> listeners = proxyEventListeners.get(proxy);
    if (listeners == null) {
      listeners = new ArrayList<ProxyEventListener>();
      proxyEventListeners.put(proxy, listeners);
    }
    listeners.add(pel);
  }

  public void notifyProxyRemoved(Object proxy) {
    List<ProxyEventListener> listeners = proxyEventListeners.get(proxy);
    if (listeners != null) {
      for (ProxyEventListener l : listeners) {
        l.proxyRemoved(proxy);
      }
    }
  }

  protected synchronized List<RMIClient> getRmiClients() {
    return new ArrayList(rmiClientMap.values());
  }

  public class TimerTask implements Runnable {
    public void run() {
      List<RMIClient> list = getRmiClients();
      for (RMIClient c : list) {
        try {
          c.renewLease();
        } catch (Exception ex) {
          ex.printStackTrace();
          // System.out.println("renew lease: "+ex.getMessage());
        }
      }
      server.removeExpiredRefs();
      server.removeServicesWithNoRemoteRefs();
    }
  }

  /**
   * Allows distributed garbage collection to close a service
   */
  public static interface Closeable {
    public void close() throws Exception;
  }

  public void close() {
    for (Client client : CLIENTS.values()) {
      client.stop();
    }
    server.close();
  }

  public static class DefaultClassAccept implements ClassAccept {
    Set<String> exclude;
    Set<String> primitives = new HashSet<String>(10);

    public DefaultClassAccept(Set<String> exclude) {
      this.exclude = exclude;
      primitives.add("int");
      primitives.add("long");
      primitives.add("double");
      primitives.add("float");
      primitives.add("byte");
      primitives.add("boolean");
      primitives.add("short");
    }

    public boolean isSpecial(String name) {
      if (primitives.contains(name) || exclude.contains(name) || name.startsWith("java.")
          || name.startsWith("org.apache.lucene.distributed") || name.startsWith("[")) {
        return false;
      }
      return true;
    }
  }

  public Server getServer() {
    return server;
  }

  // doesn't work because socket isn't made until method call on service
  public Object waitForProxy(String service, Class<?> protocol, long clientVersion, InetSocketAddress addr, boolean dgc) throws IOException {
    while (true) {
      try {
        return getProxy(service, protocol, clientVersion, addr, dgc);
      } catch (ConnectException se) { // namenode has not been started
        System.out.println("Server at " + addr + " not available yet, Zzzzz...");
      } catch (SocketTimeoutException te) { // namenode is busy
        System.out.println("Problem connecting to server: " + addr);
      }
      try {
        Thread.sleep(1000);
      } catch (InterruptedException ie) {
        // IGNORE
      }
    }
  }

  public Object getProxy(ProxyKey proxyKey) throws IOException {
    return getProxy(proxyKey.service, proxyKey.protocol, 1l, proxyKey.address, proxyKey.dgc);
  }

  public synchronized void clearOldProxies() {
    while (true) {
      Reference r = proxyRefQueue.poll();
      Object proxy = r.get();
      if (proxy != null) {
        RMIClient rmiClient = objRmiClientMap.remove(proxy);
        rmiClientMap.remove(rmiClient);
        notifyProxyRemoved(proxy);
        r.clear();
      } else
        break;
    }
  }

  // TODO: match versions
  public synchronized Object getProxy(String service, Class<?> protocol, long clientVersion, InetSocketAddress addr, boolean dgc)
      throws IOException {
    ProxyKey key = new ProxyKey(service, protocol, addr, dgc);
    RMIClient rmiClient = rmiClientMap.get(key);
    if (rmiClient == null) {
      rmiClient = getRMIClient(key);
      rmiClient.proxy = Proxy.newProxyInstance(protocol.getClassLoader(), new Class[] { protocol }, rmiClient);
      objRmiClientMap.put(rmiClient.proxy, rmiClient);
      new SoftReference(rmiClient.proxy, proxyRefQueue);
    }
    return rmiClient.proxy;
    // if (serverVersion == clientVersion) {
    // return service;
    // } else {
    // throw new VersionMismatch(protocol.getName(), clientVersion,
    // serverVersion);
    // }
  }

  /**
   * Key that identifies service proxy connection to a remote service
   * 
   */
  public static class ProxyKey implements Serializable {
    String service;
    InetSocketAddress address;
    Class<?> protocol;
    boolean dgc;

    public ProxyKey(String service, Class<?> protocol, InetSocketAddress address, boolean dgc) {
      this.service = service;
      this.protocol = protocol;
      this.address = address;
      this.dgc = dgc;
    }

    @Override
    public int hashCode() {
      final int prime = 31;
      int result = 1;
      result = prime * result + ((address == null) ? 0 : address.hashCode());
      result = prime * result + ((protocol == null) ? 0 : protocol.hashCode());
      result = prime * result + ((service == null) ? 0 : service.hashCode());
      return result;
    }

    @Override
    public boolean equals(Object obj) {
      if (this == obj)
        return true;
      if (obj == null)
        return false;
      if (getClass() != obj.getClass())
        return false;
      final ProxyKey other = (ProxyKey) obj;
      if (address == null) {
        if (other.address != null)
          return false;
      } else if (!address.equals(other.address))
        return false;
      if (protocol == null) {
        if (other.protocol != null)
          return false;
      } else if (!protocol.equals(other.protocol))
        return false;
      if (service == null) {
        if (other.service != null)
          return false;
      } else if (!service.equals(other.service))
        return false;
      return true;
    }
  }

  public synchronized RMIClient getRMIClient(ProxyKey proxyKey) throws IOException {
    // ProxyKey proxyKey = new ProxyKey(service, protocol, addr);
    RMIClient rmiClient = rmiClientMap.get(proxyKey);
    if (rmiClient == null) {
      rmiClient = new RMIClient(proxyKey, socketFactory);
      rmiClientMap.put(proxyKey, rmiClient);
    }
    return rmiClient;
  }

  private synchronized Client getClient(Configuration conf, SocketFactory factory) {
    Client client = CLIENTS.get(factory);
    if (client == null) {
      client = new Client(ByteArrayWriteable.class, conf, factory);
      CLIENTS.put(factory, client);
    }
    return client;
  }

  public class RMIClient implements InvocationHandler {
    Client client;
    InetSocketAddress address;
    String service;
    private SocketFactory factory;
    Object proxy;
    NameService nameService;
    String leaseId;
    int errors = 0;
    ProxyKey proxyKey;

    public RMIClient(ProxyKey proxyKey, SocketFactory factory) throws IOException {
      this.proxyKey = proxyKey;
      this.service = proxyKey.service;
      this.address = proxyKey.address;
      if (!service.equals("nameService"))
        this.nameService = (NameService) getProxy("nameService", NameService.class, 1l, address, false);
      this.factory = factory;
    }

    public void renewLease() {
      if (errors >= 3) {
        removeRMIClient(proxyKey);
        return;
      }
      if (leaseId != null && leaseId.equals("no"))
        return;
      if (nameService != null) {
        try {
          Lease lease = new Lease(leaseId, service, LEASE_DURATION);
          leaseId = nameService.renew(lease);
        } catch (UnknownLeaseException ule) {
          leaseId = null;
          errors++;
          renewLease();
        } catch (Exception ex) {
          ex.printStackTrace();
          errors++;
        }
      }
    }

    public Object invoke(Object proxy, Method method, Object[] params) throws Throwable {
      synchronized (this) {
        // create socket only when there is a call to the object
        if (client == null) {
          client = getClient(clientConf, factory);
          client.setTimeout(clientSocketTimeout);
        }
      }
      long startTime = System.currentTimeMillis();
      Invocation invocation = new Invocation(service, method, params);

      ByteArrayOutputStream byteOut = new ByteArrayOutputStream();
      ObjectOutputStream objectOutput = new ObjectOutputStream(byteOut);
      objectOutput.writeObject(invocation);
      objectOutput.flush();
      ByteArrayWriteable bytesSent = new ByteArrayWriteable(localhost, port, byteOut.toByteArray(), sessionId);
      if (rmiSecurity != null)
        rmiSecurity.sign(bytesSent);
      ByteArrayWriteable bytesReturned = (ByteArrayWriteable) client.call(bytesSent, address);
      if (rmiSecurity != null)
        rmiSecurity.verify(bytesReturned);
      ObjectInputStream objectInput = new ObjectInputStream(bytesReturned.getInputStream());
      // ObjectInputStream objectInput =
      // clientRmiClasses.createInput(bytesReturned.getInputStream(), RMI.this);
      Object object = null;
      try {
        object = objectInput.readObject();
      } catch (ClassNotFoundException cnfe) {
        IOException ioException = new IOException();
        ioException.initCause(cnfe);
        throw ioException;
      }
      long callTime = System.currentTimeMillis() - startTime;
      // LOG.debug("Call: " + method.getName() + " " + callTime);
      Return r = (Return) object;
      if (r.value != null)
        return r.value;
      else if (r.exception != null)
        throw r.exception;
      else
        return null;
    }
  }

  /**
   * Implementation of Writable for marshaling objects.
   * 
   */
  public static class ByteArrayWriteable implements Writable {
    public String host;
    public int port;
    public byte[] bytes;
    public byte[] signature;
    public String sessionId;

    public ByteArrayWriteable() {
    }

    public ByteArrayWriteable(String host, int port, byte[] bytes, String sessionId) {
      this.host = host;
      this.port = port;
      this.bytes = bytes;
      this.sessionId = sessionId;
    }

    public InputStream getInputStream() {
      return new ByteArrayInputStream(bytes);
    }

    public void write(DataOutput out) throws IOException {
      out.writeUTF(host);
      out.writeInt(port);
      out.writeUTF(sessionId);
      out.writeInt(bytes.length);
      out.write(bytes);
      if (signature != null) {
        out.writeBoolean(true);
        out.writeInt(signature.length);
        out.write(signature);
      } else {
        out.writeBoolean(false);
      }
    }

    public void readFields(DataInput in) throws IOException {
      host = in.readUTF();
      port = in.readInt();
      sessionId = in.readUTF();
      int length = in.readInt();
      bytes = new byte[length];
      in.readFully(bytes);
      if (in.readBoolean()) {
        signature = new byte[in.readInt()];
        in.readFully(signature);
      }
    }
  }

  public static class Invocation implements Serializable {
    public String service;
    public String methodName;
    public Object[] params;
    public Class[] paramsClasses;

    public Invocation() {
    }

    public Invocation(String service, Method method, Object[] params) {
      this.service = service;
      this.params = params;
      this.methodName = method.getName();
      this.paramsClasses = method.getParameterTypes();
    }

    public String toString() {
      StringBuilder b = new StringBuilder();
      b.append("service: " + service + " method: " + methodName);
      if (paramsClasses != null) {
        b.append(" param classes: ");
        for (int x = 0; x < paramsClasses.length; x++) {
          b.append(paramsClasses[x].getName());
          b.append(" ");
          b.append(paramsClasses[x].getClassLoader().getClass().getName());
          if (x < paramsClasses.length - 1)
            b.append(",");
        }
      }
      return b.toString();
    }
  }

  public static class Return implements Serializable {
    public Throwable exception;
    public Object value;
  }

  public static class UnknownLeaseException extends Exception {
    public UnknownLeaseException(String leaseId) {
      super(leaseId);
    }
  }

  public class Server extends org.apache.hadoop.ipc.Server {
    Map<InetSocketAddress,ClassService> classServices = new HashMap<InetSocketAddress,ClassService>();
    HashMap<String,InternalService> services = new HashMap<String,InternalService>();
    IdentityHashMap<Object,String> serviceNames = new IdentityHashMap<Object,String>();
    ReentrantLock servicesLock = new ReentrantLock();
    ReferenceQueue serviceRefQueue = new ReferenceQueue();
    Map<String,RMIServerSession> serverSessions = new HashMap<String,RMIServerSession>();

    public Server(Configuration conf, String bindAddress, int port) throws IOException {
      this(conf, bindAddress, port, 5, false);

    }

    public Server(Configuration conf, String bindAddress, int port, int numHandlers, boolean verbose) throws IOException {
      super(bindAddress, port, ByteArrayWriteable.class, numHandlers, conf);
      addService("classService", new ClassServiceImpl(), false);
      addService("nameService", new NameServiceImpl(), false);
      start();
    }

    public void close() {
      stop();
    }

    public RMIServerSession getServerSession(String sessionId, InetSocketAddress addr) throws IOException {
      synchronized (serverSessions) {
        RMIServerSession rmiServerSession = serverSessions.get(sessionId);
        if (rmiServerSession == null) {
          ClassService classService = getClassService(addr);
          rmiServerSession = new RMIServerSession(sessionId, RMI.this, classService, classAccept);
        }
        return rmiServerSession;
      }
    }

    InternalService getInternalService(String name) {
      return services.get(name);
    }

    public void removeServicesWithNoRefs() {
      servicesLock.lock();
      try {
        while (true) {
          Reference reference = serviceRefQueue.poll();
          if (reference != null) {
            Object service = reference.get();
            String name = serviceNames.remove(service);
            if (name != null) {
              services.remove(name);
            }
            reference.clear();
          } else {
            break;
          }
        }
      } finally {
        servicesLock.unlock();
      }
    }

    /**
     * Removes and closes services that are using distributed garbage collection
     * with no clients
     */
    public void removeServicesWithNoRemoteRefs() {
      servicesLock.lock();
      try {
        Iterator<InternalService> iterator = services.values().iterator();
        while (iterator.hasNext()) {
          InternalService internalService = iterator.next();
          if (internalService.getNumRefs() == 0 && internalService.dgc) {
            if (internalService.service instanceof Closeable) {
              Closeable closeable = (Closeable) internalService.service;
              try {
                closeable.close();
              } catch (Exception ex) {
                System.out.println(ex.getMessage());
              }
            }
            iterator.remove();
            System.out.println("removed dgc: " + internalService.name);
            serviceNames.remove(internalService.service);
          }
        }
      } finally {
        servicesLock.unlock();
      }
    }

    public void removeExpiredRefs() {
      servicesLock.lock();
      try {
        long now = System.currentTimeMillis();
        for (InternalService internalService : services.values()) {
          if (internalService.dgc)
            internalService.removeExpiredRefs(now);
        }
      } finally {
        servicesLock.unlock();
      }
    }

    public class InternalService {
      public String name;
      public Object service;
      public Map<String,Ref> refs = new HashMap<String,Ref>();
      public boolean dgc;

      public InternalService(String name, Object service, boolean dgc) {
        this.name = name;
        this.service = service;
        this.dgc = dgc;
        if (dgc) {
          // add default ref
          refs.put("default", new Ref(System.currentTimeMillis() + LEASE_DURATION));
        }
      }

      public void removeExpiredRefs(long now) {
        Iterator<Ref> iterator = refs.values().iterator();
        while (iterator.hasNext()) {
          Ref ref = iterator.next();
          if (now > ref.expiration)
            iterator.remove();
        }
      }

      // TODO: throw exception if lease is unknown
      public String renew(String id, long duration) throws UnknownLeaseException {
        if (!dgc)
          return "no";
        Ref ref = null;
        if (id != null)
          ref = refs.get(id);
        else {
          id = UUID.randomUUID().toString();
          ref = new Ref();
          refs.put(id, ref);
        }
        if (ref == null) {
          throw new UnknownLeaseException(id);
        }
        ref.renew(duration);
        return id;
      }

      public int getNumRefs() {
        return refs.size();
      }

      public class Ref {
        public long expiration;

        public Ref() {
        }

        public Ref(long expiration) {
          this.expiration = expiration;
        }

        public void renew(long duration) {
          expiration = System.currentTimeMillis() + duration;
        }
      }
    }

    public String renewLease(String serviceName, String leaseId, long duration) throws UnknownLeaseException {
      System.out.println("renewLease " + serviceName + " " + leaseId + " " + duration);
      InternalService is = services.get(serviceName);
      return is.renew(leaseId, duration);
    }

    public void removeService(String name) {
      servicesLock.lock();
      try {
        Object service = services.remove(name);
        serviceNames.remove(service);
      } finally {
        servicesLock.unlock();
      }
    }

    public RMI getRMI() {
      return RMI.this;
    }

    public String getServiceName(Object service) {
      servicesLock.lock();
      try {
        return serviceNames.get(service);
      } finally {
        servicesLock.unlock();
      }
    }

    public class NameServiceImpl implements NameService {
      public int getNumRefs(String service) throws Exception {
        InternalService is = services.get(service);
        if (is == null)
          throw new Exception("unknown service: " + service);
        return is.getNumRefs();
      }

      public String[] getNames() throws Exception {
        servicesLock.lock();
        try {
          String[] names = services.keySet().toArray(new String[0]);
          Arrays.sort(names);
          return names;
        } finally {
          servicesLock.unlock();
        }
      }

      public String renew(Lease lease) throws Exception {
        return renewLease(lease.service, lease.id, lease.duration);
      }
    }

    public class ClassServiceImpl implements ClassService {
      ClassLoader classLoader = Thread.currentThread().getContextClassLoader();

      public byte[] getResource(String resource) throws IOException {
        InputStream input = classLoader.getResourceAsStream(resource);
        if (input == null)
          return null;
        return IOUtils.toByteArray(input);
      }

      public byte[] loadClass(String sessionId, String className) throws Exception {
        if (sessionId.equals(RMI.this.sessionId)) {
          throw new Exception("session ids don't match");
        }
        String resource = RMIClasses.classToResource(className);
        return getResource(resource);
      }
    }

    public void addService(String name, Object service, boolean dgc) {
      servicesLock.lock();
      try {
        services.put(name, new InternalService(name, service, dgc));
        serviceNames.put(service, name);
        new SoftReference(service, serviceRefQueue);
      } finally {
        servicesLock.unlock();
      }
    }

    public ClassService getClassService(InetSocketAddress addr) throws IOException {
      return (ClassService) getProxy("classService", ClassService.class, ClassService.serialVersionUID, addr, false);
    }

    protected ByteArrayWriteable toByteArrayWriteable(Object o) throws IOException {
      try {
        ByteArrayOutputStream byteOut = new ByteArrayOutputStream();
        ObjectOutputStream objectOutput = new RMIServerObjectOutputStream(byteOut, this);
        objectOutput.writeObject(o);
        objectOutput.flush();
        ByteArrayWriteable baw = new ByteArrayWriteable(localhost, port, byteOut.toByteArray(), sessionId);
        if (rmiSecurity != null) {
          rmiSecurity.sign(baw);
        }
        return baw;
      } catch (IOException ioException) {
        throw ioException;
      } catch (Exception e) {
        IOException ioe = new IOException(e.toString());
        ioe.setStackTrace(e.getStackTrace());
        throw ioe;
      }
    }

    public Writable call(Writable param) throws IOException {
      try {
        ByteArrayWriteable bytes = (ByteArrayWriteable) param;
        if (rmiSecurity != null)
          rmiSecurity.verify(bytes);
        InetSocketAddress addr = new InetSocketAddress(bytes.host, bytes.port);
        RMIServerSession serverSession = getServerSession(bytes.sessionId, addr);
        ObjectInputStream objectIn = serverSession.getObjectInputStream(new ByteArrayInputStream(bytes.bytes));
        Invocation invocation = (Invocation) objectIn.readObject();
        InternalService is = services.get(invocation.service);
        System.out.println("invocation.service: " + invocation.service + " class: " + is.service.getClass().getName() + "."
            + invocation.methodName);
        Method method = is.service.getClass().getMethod(invocation.methodName, invocation.paramsClasses);
        long startTime = System.currentTimeMillis();
        Object value = method.invoke(is.service, invocation.params);
        Return r = new Return();
        r.value = value;
        ByteArrayWriteable outBytes = toByteArrayWriteable(r);
        if (rmiSecurity != null)
          rmiSecurity.sign(outBytes);
        return outBytes;
      } catch (InvocationTargetException e) {
        Throwable target = e.getTargetException();
        Return r = new Return();
        r.exception = target;
        return toByteArrayWriteable(r);
      } catch (Throwable e) {
        IOException ioe = new IOException(e.toString());
        ioe.setStackTrace(e.getStackTrace());
        throw ioe;
      }
    }
  }
}
